package io.github.wanggit.antrpc.client.monitor.report;

import com.google.common.collect.Lists;
import io.github.wanggit.antrpc.commons.bean.IdGenHelper;
import io.github.wanggit.antrpc.commons.bean.RpcCallLog;
import io.github.wanggit.antrpc.commons.config.IConfiguration;
import io.github.wanggit.antrpc.commons.utils.JsonUtils;
import io.github.wanggit.antrpc.commons.utils.LongToDateUtil;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import zipkin2.Endpoint;
import zipkin2.Span;

import java.util.Map;

public class ZipkinReporter extends AbstractLogReporter {

    public ZipkinReporter(IConfiguration configuration, ApplicationContext context) {
        super(configuration);
        initKafkaTracingSender(configuration, context);
    }

    private void initKafkaTracingSender(IConfiguration configuration, ApplicationContext context) {
        KafkaProperties properties = context.getBean(KafkaProperties.class);
        Map<String, Object> props = properties.getProducer().buildProperties();
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
        DefaultKafkaProducerFactory<?, ?> factory = new DefaultKafkaProducerFactory<>(props);
        String transactionIdPrefix = properties.getProducer().getTransactionIdPrefix();
        if (transactionIdPrefix != null) {
            factory.setTransactionIdPrefix(transactionIdPrefix);
        }
        KafkaTracingSender.builder()
                .setTopic(configuration.getCallLogReporterConfig().getZipkinTopic())
                .setKafkaTemplate(new KafkaTemplate<>(factory))
                .setLocalEndpoint(
                        Endpoint.newBuilder()
                                .ip(configuration.getExposeIp())
                                .port(configuration.getPort())
                                .serviceName(configuration.getApplicationName())
                                .build())
                .setReportAllMethodsToZipkin(
                        configuration.getCallLogReporterConfig().isReportAllMethodsToZipkin())
                .build();
    }

    @Override
    protected void internalReport(
            RpcCallLog log, ZipkinAllMethodTracing.TracingPairs lastTracingPairs) {
        String parentId = IdGenHelper.to16LowHexZipkinId(log.getCaller());
        parentId =
                IdGenHelper.isZipkinIdAllZero(parentId) && null != lastTracingPairs
                        ? IdGenHelper.to16LowHexZipkinId(lastTracingPairs.getRequestId())
                        : parentId;
        Span.Kind kind =
                IdGenHelper.isZipkinIdAllZero(parentId) ? Span.Kind.CLIENT : Span.Kind.SERVER;
        Span span =
                Span.newBuilder()
                        .timestamp(log.getStart() * 1000)
                        .localEndpoint(
                                Endpoint.newBuilder()
                                        .ip(log.getIp())
                                        .port(log.getPort())
                                        .serviceName(log.getAppName())
                                        .build())
                        .remoteEndpoint(
                                Endpoint.newBuilder()
                                        .ip(log.getTargetIp())
                                        .port(log.getTargetPort())
                                        .serviceName(log.getTargetAppName())
                                        .build())
                        .name(log.getTargetMethodName())
                        .kind(kind)
                        .duration(log.getRt() * 1000)
                        .parentId(parentId)
                        .id(IdGenHelper.to16LowHexZipkinId(log.getRequestId()))
                        .traceId(log.getSerialNumber())
                        .putTag(
                                "0_client",
                                log.getAppName() + "@" + log.getIp() + ":" + log.getPort())
                        .putTag(
                                "1_targetServer",
                                log.getTargetAppName()
                                        + "@"
                                        + log.getTargetIp()
                                        + ":"
                                        + log.getTargetPort())
                        .putTag(
                                "2_previousStackTrace",
                                (null == log.getPreviousStackTrace()
                                        ? ""
                                        : log.getPreviousStackTrace()))
                        .putTag("3_targetClassName", log.getTargetClassName())
                        .putTag("4_targetMethodName", log.getTargetMethodName())
                        .putTag(
                                "5_requestArgs",
                                (null == log.getRequestArgs())
                                        ? ""
                                        : JsonUtils.toJsonString(log.getRequestArgs()))
                        .putTag("6_start", LongToDateUtil.toDateMsStr(log.getStart(), ""))
                        .putTag("7_rt(ms)", String.valueOf(log.getRt()))
                        .putTag("8_end", LongToDateUtil.toDateMsStr(log.getEnd(), ""))
                        .putTag(
                                "9_errorMessage",
                                (null == log.getErrorMessage() ? "" : log.getErrorMessage()))
                        .putTag("a_requestId", log.getRequestId())
                        .putTag("b_parentId", log.getCaller())
                        .putTag("c_threadId", String.valueOf(log.getThreadId()))
                        .putTag("d_date", log.getDate())
                        .build();
        KafkaTracingSender.getInstance().trace(Lists.newArrayList(span));
    }
}
